package com.atguigu.gmall.app.function;

import com.alibaba.druid.pool.DruidDataSource;
import com.alibaba.druid.pool.DruidPooledConnection;
import com.alibaba.fastjson.JSONObject;
import com.atguigu.gmall.util.DimUtil;
import com.atguigu.gmall.util.DruidDSUtils;
import com.atguigu.gmall.util.PhoenixUtils;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.streaming.api.functions.sink.RichSinkFunction;

/**
 * @author : ranzlupup
 * @since : 2023/6/2 08:59
 */
public class DimSinkFunction extends RichSinkFunction<JSONObject> {

    // Druid连接池
    private DruidDataSource druidDataSource = null;

    @Override
    public void open(Configuration parameters) throws Exception {
        druidDataSource = DruidDSUtils.createDataSource();
    }


    /**
     * 每条数据调用一次invoke
     * 每条数据是经过处理后的主流数据
     *
     * @param value   //value:{"database":"gmall-211126-flink","table":"base_trademark","type":"update","ts":1652499176,"xid":188,"commit":true,"data":{"id":13,"tm_name":"atguigu"},"old":{"logo_url":"/aaa/aaa"},"sinkTable":"dim_xxx"}
     *                //value:{"database":"gmall-211126-flink","table":"order_info","type":"update","ts":1652499176,"xid":188,"commit":true,"data":{"id":13,...},"old":{"xxx":"/aaa/aaa"},"sinkTable":"dim_xxx"}
     * @param context 上下文信息
     */
    @Override
    public void invoke(JSONObject value, Context context) throws Exception {
        // 获取连接
        DruidPooledConnection connection = druidDataSource.getConnection();

        // Phoenix中的表名
        String sinkTable = value.getString("sinkTable");
        JSONObject data = value.getJSONObject("data");

        // 获取数据类型
        String type = value.getString("type");

        //! 如果为更新数据，则需要删除redis中的维度数据
        if ("update".equals(type)) {
            DimUtil.delDimInfo(sinkTable.toUpperCase(), data.getString("id"));
        }


        // 写出数据
        PhoenixUtils.upsertValues(connection, sinkTable, data);

        //归还连接
        connection.close();
    }

    // @Override
    // public void close() throws Exception {
    //     druidDataSource.close();
    // }
}
